package com.sptus.framework.kafka;

import org.apache.kafka.clients.producer.ProducerRecord;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;

import lombok.extern.slf4j.Slf4j;

import org.springframework.beans.factory.annotation.Autowired;
import javax.annotation.PostConstruct;

@Component
@Slf4j
public class KafkaProducerService {

	// ✅ 静态变量存储 KafkaTemplate
	private static KafkaTemplate<String, String> kafkaTemplateStatic;

	@Autowired
	private KafkaTemplate<String, String> kafkaTemplate;

	/**
	 * ✅ 初始化静态 KafkaTemplate（通过 Spring 注入后自动赋值）
	 */
	@PostConstruct
	public void init() {
		KafkaProducerService.kafkaTemplateStatic = this.kafkaTemplate;
	}

	/**
	 * ✅ 静态方法：发送消息到 Kafka 使用方式：KafkaProducerService.sendMessage("device-topic",
	 * "SN123456", "{\"SN\":\"SN123456\", \"MSG\":\"0\"}");
	 *
	 * @param topic   Kafka 主题
	 * @param key     消息的 key（可为 null）
	 * @param message 消息内容（value）
	 */
	public static void sendMessage(String topic, String key, String message) {
		if (kafkaTemplateStatic == null) {
			System.err.println("❌ KafkaTemplate 未初始化，请检查是否已正确注入");
			return;
		}
		try {
			ProducerRecord<String, String> record = new ProducerRecord<>(topic, key, message);
			kafkaTemplateStatic.send(record);
			log.info("✅ 消息已发送到 Kafka topic: " + topic + ", key: " + key);
		} catch (Exception e) {
			log.error("❌ Kafka 消息发送失败：" + e.getMessage());
			e.printStackTrace();
		}
	}

	/**
	 * ✅ 快捷方法：无 key 的消息发送
	 */
	public static void sendMessage(String topic, String message) {
		sendMessage(topic, null, message);
	}
}